package main;

import java.io.IOException;

import mapreduce.AssociationRulesMapper;
import mapreduce.AssociationRulesReducer;

import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.TextInputFormat;
import org.apache.hadoop.mapred.TextOutputFormat;
import org.apache.hadoop.mapreduce.Job;

import util.CharacterSet;

public class Main {

	private static final String RUNNING_JAR = "mapred.jar";
	private static final int DEFAULT_SLICE_SIZE = 1000;

	public Main(long minOcurrences, int size, int sliceSize, String inputPath, String outputPath) throws IOException, ClassNotFoundException, InterruptedException{
		System.out.println("Configuring");
		JobConf conf = new JobConf();
		conf.setLong("MIN_OCCURRENCES", minOcurrences);
		conf.setInt("SIZE", size);
		conf.setInt("SLICE_SIZE", sliceSize);

		// here you have to put your mapper class
		conf.setMapperClass(AssociationRulesMapper.class);
		// here you have to put your reducer class
		conf.setReducerClass(AssociationRulesReducer.class);
		// here you have to set the jar which is containing your 
		// map/reduce class, so you can use the mapper class
		conf.setJar(RUNNING_JAR);
		conf.setJarByClass(Main.class);
		// key/value of the mapper output
		conf.setMapOutputKeyClass(CharacterSet.class);
		conf.setMapOutputValueClass(LongWritable.class);

		// key/value of the reducer output
		conf.setOutputKeyClass(CharacterSet.class);
		conf.setOutputValueClass(LongWritable.class);
		// this is setting the format of your input, can be TextInputFormat
		conf.setInputFormat(TextInputFormat.class);
		// same with output
		conf.setOutputFormat(TextOutputFormat.class);			
		// create a new job based on the configuration
		// here you can set the path of your input
		TextInputFormat.addInputPath(conf, new Path(inputPath));
		FileSystem fs = FileSystem.get(conf);
		// this deletes possible output paths to prevent job failures
		Path out = new Path(outputPath);
		fs.delete(out, true);
		// finally set the empty out path
		TextOutputFormat.setOutputPath(conf, out);

		Job job = new Job(conf);
		// this waits until the job completes and prints debug out to STDOUT or whatever
		// has been configured in your log4j properties.
		System.out.println("Started job");
		job.waitForCompletion(true);
		System.out.println("Finished job");
	}

	public static void main(String[] args) throws ClassNotFoundException, IOException, InterruptedException {
		long minOccurrences = Long.parseLong(args[0]);
		int size = Integer.parseInt(args[1]);
		new Main(minOccurrences, size, DEFAULT_SLICE_SIZE, args[2], args[3]);
	}
}
